package com.aaron.remoting.transport.netty.client;


import com.aaron.enums.CompressTypeEnum;
import com.aaron.enums.SerializationTypeEnum;
import com.aaron.extension.ExtensionLoader;
import com.aaron.factory.SingletonFactory;
import com.aaron.registry.ServiceDiscovery;
import com.aaron.remoting.constants.RpcConstants;
import com.aaron.remoting.dto.RpcMessage;
import com.aaron.remoting.dto.RpcRequest;
import com.aaron.remoting.dto.RpcResponse;
import com.aaron.remoting.transport.RpcRequestTransport;
import com.aaron.remoting.transport.netty.codec.RpcMessageDecoder;
import com.aaron.remoting.transport.netty.codec.RpcMessageEncoder;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;


@Slf4j
public final class NettyRpcClient implements RpcRequestTransport {
    // 服务发现者，通过服务名，返回服务地址
    private final ServiceDiscovery serviceDiscovery;
    // 未处理的Request请求
    private final UnprocessedRequests unprocessedRequests;
    // 存储Channel管道
    private final ChannelProvider channelProvider;
    // 启动类
    private final Bootstrap bootstrap;
    // 事件循环组
    private final EventLoopGroup eventLoopGroup;

    public NettyRpcClient() {
        // 初始化netty的EventLoopGroup, Bootstrap
        eventLoopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.INFO))
                // 连接的建立时间，如果超过此时间无法建立连接则失败。5秒
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline p = ch.pipeline();
                        // 如果15秒没有发送消息，那么就发送一次心跳。保持连接。写操作的超时时间为5秒
                        p.addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
                        // 以下三个就是出入站的handler
                        p.addLast(new RpcMessageEncoder()); // 出站需要编码Encode，它里面有形参RpcMessage
                        p.addLast(new RpcMessageDecoder()); // 入站需要解码Decode
                        p.addLast(new NettyRpcClientHandler()); // 业务处理的handler
                    }
                });
        this.serviceDiscovery = ExtensionLoader.getExtensionLoader(ServiceDiscovery.class).getExtension("zk");
        // 反射获取实例化对象
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
        // 反射获取实例化对象
        this.channelProvider = SingletonFactory.getInstance(ChannelProvider.class);
    }

    /**
     * 连接服务器并获取通道
     * 通过套接字接口，返回Channel管道
     */
    @SneakyThrows
    public Channel doConnect(InetSocketAddress inetSocketAddress) {
        // completableFuture
        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
        // future-listener监听机制
        bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.info("The client has connected [{}] successful!", inetSocketAddress.toString());
                // 完成异步执行，异步就可以拿到Channel管道
                completableFuture.complete(future.channel());
            } else {
                throw new IllegalStateException();
            }
        });
        // 这里就是上面异步拿到的Channel管道
        return completableFuture.get();
    }


    // 传输rpc请求（RpcRequest）到服务端,这里是代理类RpcClientProxy调用这个方法
    @Override
    public Object sendRpcRequest(RpcRequest rpcRequest) {
        // 使用CompletableFuture包装返回值
        CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();
        // 通过zk，返回套接字地址
        InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest);
        // 通过套接字地址，找到通道
        Channel channel = getChannel(inetSocketAddress);

        if (channel.isActive()) {
            // 放入未处理的请求. 实际是放入一个Map容器中，key为requestId，value为CompletableFuture<>。
            // 这里就是在等待 异步请求的通知。当NettyRpcClientHandler里面的complete方法调用，后面就可以使用CompletableFuture.get()去拿了
            unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture);
            // 给RpcMessage配置其他信息：data、codec、compress、messageType
            RpcMessage rpcMessage = RpcMessage.builder().data(rpcRequest)
                    .codec(SerializationTypeEnum.PROTOSTUFF.getCode())
                    .compress(CompressTypeEnum.GZIP.getCode())
                    .messageType(RpcConstants.REQUEST_TYPE).build();
            // 将rpcMessage写入管道，并设置future-listener机制.这个里面传输了我们出站时需要使用RpcMessageEncoder进行编码的形参
            channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("client send message: [{}]", rpcMessage);
                } else {
                    // 传输失败
                    future.channel().close();
                    // 发送错误处理
                    resultFuture.completeExceptionally(future.cause());
                    log.error("Send failed:", future.cause());
                }
            });
        } else {
            throw new IllegalStateException();
        }
        // 返回的是 CompletableFuture<RpcResponse<Object>>
        return resultFuture;
    }

    // 根据InetSocketAddress获取Channel(从channelProvider的ChannelMap中),还是要调用上面的doConnect方法
    public Channel getChannel(InetSocketAddress inetSocketAddress) {
        Channel channel = channelProvider.get(inetSocketAddress);
        if (channel == null) {
            // 连接并获取channel
            channel = doConnect(inetSocketAddress);
            channelProvider.set(inetSocketAddress, channel);
        }
        return channel;
    }

    // 关闭eventLoopGroup
    public void close() {
        eventLoopGroup.shutdownGracefully();
    }
}
